package com.zfei.firstflink;

import com.zfei.firstflink.model.XexHomeOrder;
import com.zfei.firstflink.utils.DataSetMysqlSource;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import scala.Tuple2;
import scala.Tuple3;
import scala.Tuple4;

import java.math.BigDecimal;

/***
 *$com.zfei.firstflink.TestDataset
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/22:下午2:23
 *
 *description:
 */
public class TestDataset {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();
        DataSet<XexHomeOrder> orderDateSet= env.createInput(
                new DataSetMysqlSource()
//                JDBCInputFormat.buildJDBCInputFormat()
//                .setDrivername("com.mysql.jdbc.Driver").setDBUrl("jdbc:mysql://47.100.10.9:3306/civil_admin")
//                .setPassword("Zkxbx-2011!").setUsername("zkxbxmysql").setQuery("" +
//                        "select order_no,service_object_id,service_provider_id,state,order_type,number,price,spotprice,order_time,business_time,start_time,end_time from civil_admin.xex_home_order order by order_time asc").setRowTypeInfo(new RowTypeInfo()).finish()
        );

//        orderDateSet.flatMap(new FlatMapFunction<Row, XexHomeOrder>() {
//            @Override
//            public void flatMap(Row row, Collector<XexHomeOrder> collector) throws Exception {
//               try {
//                   XexHomeOrder order=new XexHomeOrder();
//                   order.setOrderNo(row.getField(1).toString());
//                   order.setServiceObjectId(row.getField(2).toString());
//                   order.setServiceProviderId(row.getField(3).toString());
//                   order.setState(Integer.parseInt(row.getField(4).toString()));
//                   order.setOrderType(Integer.parseInt(row.getField(5).toString()));
//                   order.setNumber(row.getField(6).toString());
//                   order.setPrice(BigDecimal.valueOf(Double.parseDouble(row.getField(7).toString())));
//                   collector.collect(order);
//
//               }catch (Exception ex){
//
//               }
//            }


   orderDateSet.filter(item->item.getOrderType()==4).flatMap(new FlatMapFunction<XexHomeOrder, Tuple4<Integer,Integer,BigDecimal,XexHomeOrder>>() {
       @Override
       public void flatMap(XexHomeOrder xexHomeOrder, Collector<Tuple4<Integer,Integer,BigDecimal,XexHomeOrder>> collector) throws Exception {
           if (xexHomeOrder==null) return;
          collector.collect(new  Tuple4<Integer,Integer,BigDecimal,XexHomeOrder>(xexHomeOrder.getState(),1,xexHomeOrder.getPrice(),xexHomeOrder));
       }
   }).groupBy(item->item._1()).reduce(new ReduceFunction<Tuple4<Integer,Integer,BigDecimal,XexHomeOrder>>() {
       @Override
       public Tuple4<Integer, Integer, BigDecimal, XexHomeOrder> reduce(Tuple4<Integer, Integer, BigDecimal, XexHomeOrder> result, Tuple4<Integer, Integer, BigDecimal, XexHomeOrder> t1) throws Exception {

           Tuple4 t4=new Tuple4<Integer, Integer, BigDecimal, XexHomeOrder>(result._1(),result._2()+1,result._3().add(t1._3()),null);
         return t4;
       }
   })
//           .flatMap(new FlatMapFunction<Tuple4<Integer,Integer,BigDecimal,XexHomeOrder>, Tuple2<Integer,BigDecimal>>() {
//       @Override
//       public void flatMap(Tuple4<Integer, Integer, BigDecimal, XexHomeOrder> t4, Collector<Tuple2<Integer, BigDecimal>> collector) throws Exception {
//           if ()
//           collector.collect(new Tuple2<>(t4._1(),t4._3().divide(BigDecimal.valueOf(t4._2()))));
//       }
//   })
        .print();
          ;
      // env.execute();
    }
}
